package com.atguigu.flink08;

import com.atguigu.beans.WaterSensor;
import com.atguigu.func.WaterSensorMapFunction;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author Felix
 * @date 2024/2/27
 * 该案例演示了topn的第二种实现思路
 */
public class Flink01_TopN_2 {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(1);
        //TODO 2.从指定的网络端口中读取数据
        DataStreamSource<String> socketDS = env.socketTextStream("hadoop102", 8888);
        //TODO 3.对读取的数据进行类型的转换 String -> Watersensor
        SingleOutputStreamOperator<WaterSensor> wsDS = socketDS.map(new WaterSensorMapFunction());
        //TODO 4.指定Watermark以及提取事件时间字段
        SingleOutputStreamOperator<WaterSensor> withWatermarkDS = wsDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<WaterSensor>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<WaterSensor>() {
                                    @Override
                                    public long extractTimestamp(WaterSensor ws, long recordTimestamp) {
                                        return ws.ts * 1000;
                                    }
                                }
                        )
        );
        //TODO 5.按照水位值进行分组
        KeyedStream<WaterSensor, Integer> vcKeyedDS = withWatermarkDS.keyBy(WaterSensor::getVc);
        //TODO 6.对分组后的数据进行开窗
        WindowedStream<WaterSensor, Integer, TimeWindow> windowDS = vcKeyedDS.window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)));

        //TODO 7.对窗口中的数据进行处理  增量:统计水位值出现的次数 + 全量：补充窗口信息 ---->Tuple3<VC,Count,End>
        SingleOutputStreamOperator<Tuple3<Integer, Integer, Long>> aggregateDS = windowDS.aggregate(
                new MyAgg(),
                new MyProcess()
        );
        //aggregateDS.print();

        //TODO 8.按照窗口的结束时间进行分组
        KeyedStream<Tuple3<Integer, Integer, Long>, Long> windowEndKeyedDS = aggregateDS.keyBy(t -> t.f2);

        //TODO 9.对分组后的数据进行处理
        SingleOutputStreamOperator<String> processDS = windowEndKeyedDS.process(
                new MyTopn()
        );

        processDS.print();

        env.execute();
    }
}

class MyTopn extends KeyedProcessFunction<Long, Tuple3<Integer, Integer, Long>, String>{
    //定义一个map集合，用于存放每个窗口，水位值以及出现次数
    private Map<Long, List<Tuple3<Integer, Integer, Long>>> vcCountMap = new HashMap<>();
    @Override
    public void processElement(Tuple3<Integer, Integer, Long> vcCountEnd, KeyedProcessFunction<Long, Tuple3<Integer, Integer, Long>, String>.Context ctx, Collector<String> out) throws Exception {
        Long windowEnd = ctx.getCurrentKey();
        //判断map集合中是否包含当前窗口
        if(vcCountMap.containsKey(windowEnd)){
            //说明窗口中已经有数据
            vcCountMap.get(windowEnd).add(vcCountEnd);
        }else{
            //说明窗口中还没有数据
            List<Tuple3<Integer, Integer, Long>> vcCountList = new ArrayList<>();
            vcCountList.add(vcCountEnd);
            vcCountMap.put(windowEnd,vcCountList);
        }

        //为了能够接收到上游当前窗口的所有数据，我们这里注册一个事件时间定时器，定时器触发的时候，再排序
        TimerService timerService = ctx.timerService();
        timerService.registerEventTimeTimer(windowEnd + 1);
    }

    //定时器被触发的时候，执行的方法
    @Override
    public void onTimer(long timestamp, KeyedProcessFunction<Long, Tuple3<Integer, Integer, Long>, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
        Long windowEnd = ctx.getCurrentKey();
        //从维护的map集合中，将当前窗口中的水位值以及出现的次数取出来
        List<Tuple3<Integer, Integer, Long>> vcCountEndList = vcCountMap.get(windowEnd);
        //对list集合中的元素进行排序
        vcCountEndList.sort((o1, o2) -> o2.f1 - o1.f1);

        //从集合中取出数据
        StringBuilder outStr = new StringBuilder();

        outStr.append("================================\n");
        // 遍历 排序后的 List，取出前 threshold 个， 考虑可能List不够2个的情况  ==》 List中元素的个数 和 2 取最小值
        for (int i = 0; i < Math.min(2, vcCountEndList.size()); i++) {
            Tuple3<Integer, Integer, Long> vcCount = vcCountEndList.get(i);
            outStr.append("Top" + (i + 1) + "\n");
            outStr.append("vc=" + vcCount.f0 + "\n");
            outStr.append("count=" + vcCount.f1 + "\n");
            outStr.append("窗口结束时间=" + vcCount.f2 + "\n");
            outStr.append("================================\n");
        }

        vcCountEndList.clear();
        //将数据传递到下游去
        out.collect(outStr.toString());
    }
}
class MyProcess extends ProcessWindowFunction<Integer, Tuple3<Integer,Integer,Long>, Integer, TimeWindow>{
    @Override
    public void process(Integer key, ProcessWindowFunction<Integer, Tuple3<Integer, Integer, Long>, Integer, TimeWindow>.Context context, Iterable<Integer> elements, Collector<Tuple3<Integer, Integer, Long>> out) throws Exception {
        Integer count = elements.iterator().next();
        long end = context.window().getEnd();
        out.collect(Tuple3.of(key,count,end));
    }
}
class MyAgg implements AggregateFunction<WaterSensor, Integer, Integer>{

    @Override
    public Integer createAccumulator() {
        return 0;
    }

    @Override
    public Integer add(WaterSensor value, Integer accumulator) {
        return accumulator + 1;
    }

    @Override
    public Integer getResult(Integer accumulator) {
        return accumulator;
    }

    @Override
    public Integer merge(Integer a, Integer b) {
        return null;
    }
}
